package com.coolpad.job.device.service.spider;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.coolpad.job.device.dao.GeoInfoDao;
import com.coolpad.job.device.dao.HBaseDao;
import com.coolpad.job.device.model.CityPartitionPO;
import com.coolpad.job.device.model.GeoFullInfoPO;
import com.coolpad.job.device.model.Point;
import com.coolpad.job.device.util.Function.FunctionUtils;
import com.coolpad.job.device.util.RailScanUtils;
import com.jayway.jsonpath.JsonPath;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.ListUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.converter.StringHttpMessageConverter;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;
import org.springframework.web.client.RestTemplate;

import java.io.BufferedReader;
import java.io.IOException;
import java.nio.charset.Charset;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.LinkedHashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;

@Slf4j
//@Service
public class GeoSpiderService {

    @Autowired
    private RestTemplate restTemplate;

    @Autowired
    private GeoInfoDao geoInfoDao;

    @Autowired
    private HBaseDao hBaseDao;

    private List<String> keys = Arrays.asList("20a62501e82c73ebdb969184a27cf613");
    private String currentKey = "";
    private int keyIndex = 0;
    private String geo_url_temp = "https://restapi.amap.com/v3/geocode/regeo?location=%s&key=%s&extensions=base&batch=true&roadlevel=0";
    private String rail_url_temp = "https://restapi.amap.com/v3/config/district?key=%s&keywords=%s&subdistrict=0&extensions=all";
    public static List<String> failCodes = new ArrayList<>();
    private AtomicLong adder = new AtomicLong(0);
    private ExecutorService executor;

    {
        exchangeKey();
        ThreadFactory threadFactory = new CustomizableThreadFactory("geo-pool-");
        executor = new ThreadPoolExecutor(10, 10,
                0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(),threadFactory);
    }


    public String exchangeKey(){
        currentKey = keys.get(keyIndex++);

        return currentKey;
    }

    /**
     * 加载所有区域的点
     */
    public void loadAllAdPoint(){

        //广东省
        String provinceName = "广东省";
        List<String> adCodes = readAdCode(provinceName);
        for (int j = 0; j < adCodes.size(); j++) {
            String countyInfo = adCodes.get(j);
            String[] split = countyInfo.split("#");
            String adCode = split[0];
            try {
                String rail = FunctionUtils.retryIfException(adCode.substring(0,6),3, code -> {
                    log.info("请求 adCode:{} 围栏",code);
                    return requestRail(code);
                });

                // scan rail
                if(StringUtils.hasText(rail)){

                    List<Point> points = RailScanUtils.scanPoint(rail,adCode);
                    List<List<Point>> partition = ListUtils.partition(points, 6000);

                    log.info("adCode:{} 分批:{} 插入开始", adCode, partition.size());
                    for (int i = 0; i < partition.size(); i++){

                        List<Point> part = partition.get(i);
                        FunctionUtils.retryIfException(3, () -> {
                            int res = geoInfoDao.batchInsert(part);
                            log.info("批量插入成功 :{} ",res);
                        });
                        log.info("adCode[{}/{}]:{} 插入进度:{}/{}", j+1, adCodes.size(), adCode, i+1, partition.size());
                    }
                    adder.addAndGet(points.size());
                }
                log.info("adCode:{}]  打点完成",adder.get());
            }catch (Exception e) {
                log.error("[adCode:{}] 打点异常", adCode);
            }
        }
        log.info("{}所有adCode打点完成：{}",provinceName,adder.get());
    }

    /**
     * 加载所有区域的点
     */
    public void scanPointToHBase(String provinceName){

        //广东省
//        String provinceName = "广东省";
        List<String> adCodes = geoInfoDao.findAdCode(provinceName);
        for (int j = 0; j < adCodes.size(); j++) {
            String adCode = adCodes.get(j);
            scanAdToHBase(adCode);
        }
        log.info("{}所有adCode打点完成：{}",provinceName,adder.get());
    }

    private void scanAdToHBase(String adCode) {
        try {
            String rail = FunctionUtils.retryIfException(adCode.substring(0,6),3, code -> {
                log.info("请求 adCode:{} 围栏",code);
                return requestRail(code);
            });

            // scan rail
            if(StringUtils.hasText(rail)){
                List<Point> points = RailScanUtils.scanPoint(rail,adCode);
                List<List<Point>> partitions = ListUtils.partition(points, 20);
                log.info("扫描 part num :{}",points.size());
                for(int i = 0; i < partitions.size(); i++){
                    List<Point> part = partitions.get(i);
                    executor.submit(() -> requestGeo2HBase(part));
                    log.info("submit progress[{}/{}]",i,partitions.size());
                }
                while (executor.isTerminated()){
                    hBaseDao.flush();
                }
                log.info("adCode:{} put to HBase buffer finish",adCode);
            }
            log.info("adCode:[{}]  打点完成",adder.get());
        }catch (Exception e) {
            log.error("[adCode:{}] 打点异常", adCode);
        }
    }

    /**
     * 请求逆地理
     * @param points
     */
    public void requestGeo2HBase(List<Point> points) {

        String locations = points.stream()
                .map(point -> String.format("%s,%s"))
                .collect(Collectors.joining("|"));
        restTemplate.getMessageConverters().set(1, new StringHttpMessageConverter(Charset.forName("utf-8")));
        String response = restTemplate.getForEntity(String.format(geo_url_temp, locations, currentKey), String.class).getBody();
        String code = JsonPath.read(response, "$.infocode");
        if("10003".equals(code)) {
            response = restTemplate.getForEntity(String.format(geo_url_temp, locations, exchangeKey()), String.class).getBody();
        }
        if("10000".equals(code)){
            List<GeoFullInfoPO> pos = convert2GeoInfo(points, response);
            boolean putCount = hBaseDao.put(pos);
            log.info("put to HBase buffer count:{}",putCount);
        }
    }

    /**
     * 转换为po
     * @param response
     * @return
     */
    private List<GeoFullInfoPO> convert2GeoInfo(List<Point> points, String response) {

        JSONArray reGeoCodes = JSONObject.parseObject(response).getJSONArray("regeocodes");
        List<GeoFullInfoPO> pos = new ArrayList<>(20);
        if(Objects.isNull(reGeoCodes)){
            return Collections.emptyList();
        }
        for(int i = 0; i < reGeoCodes.size(); i++){
            JSONObject jsonObject = reGeoCodes.getJSONObject(i);
            GeoFullInfoPO po = GeoFullInfoPO.builder()
                    .lon(String.valueOf(points.get(i).getX()))
                    .lat(String.valueOf(points.get(i).getY()))
                    .provinceName(jsonObject.getString("province"))
                    .cityId(jsonObject.getLongValue("citycode"))
                    .cityName(jsonObject.getString("city"))
                    .townId(jsonObject.getLongValue("township"))
                    .townName(jsonObject.getString("towncode"))
                    .build();
            pos.add(po);
        }
        return pos;
    }

    public String requestRail(String keywords){
        String rail = "";
        String url = String.format(rail_url_temp, keys.get(0), keywords);
        restTemplate.getMessageConverters().set(1, new StringHttpMessageConverter(Charset.forName("utf-8")));
        String response = restTemplate.getForEntity(url, String.class).getBody();
        try {
            if("1".equals(JsonPath.read(response,"$.status"))) {
                if(keywords.equals(JsonPath.read(response,"$.districts[0].adcode"))){
                    rail = JsonPath.read(response,"$.districts[0].polyline");
                }
            }
        }catch (Exception e){
            failCodes.add(keywords);
            log.error("find {} rail error ",keywords, e);
        }
        return rail;
    }

    /**
     * 读取第三级城乡划分id
     * @return
     */
    public List<String> readAdCode() {

        BufferedReader reader = null;
        try {
            reader = Files.newBufferedReader(Paths.get("C:\\Users\\coolpad\\Desktop\\数据采集\\city_table.txt"), Charset.forName("utf-8"));
        } catch (IOException e) {
            log.error("",e);
        }
        LinkedHashSet<String> adSet = new LinkedHashSet<>();
        reader.lines()
                .forEach(line -> {
                    if(StringUtils.hasText(line)){
                        String[] lineArr = line.split(",");
                        adSet.add(lineArr[2]);
                    }
                });
        adSet.remove("440305000000");
        return new ArrayList<>(adSet);
    }

    /**
     * 读取第三级城乡划分id
     * @return
     */
    public List<String> readAdCode(String provinceName) {

        List<String> adCode = geoInfoDao.findAdCode(provinceName);
        return adCode;
    }

    /**
     * 读取第三级城乡划分id
     * @return
     */
    public List<CityPartitionPO> readCity() {

        AtomicLong adder = new AtomicLong(10000);
        BufferedReader reader = null;
        try {
            reader = Files.newBufferedReader(Paths.get("C:\\Users\\coolpad\\Desktop\\数据采集\\city_table.txt"), Charset.forName("utf-8"));
        } catch (IOException e) {
            log.error("",e);
        }
        List<CityPartitionPO> poList = new LinkedList<>();
        reader.lines()
                .forEach(line -> {
                    if(StringUtils.hasText(line)){
                        String[] lineArr = line.split(",");
                        CityPartitionPO po = new CityPartitionPO();
                        po.setPartitionId(adder.addAndGet(1));
                        po.setTownId(Long.valueOf(lineArr[0]));
                        po.setTownName(lineArr[1]);
                        po.setCountyId(Long.valueOf(lineArr[2]));
                        po.setCountyName(lineArr[3]);
                        po.setCityId(Long.valueOf(lineArr[4]));
                        po.setCityName(lineArr[5]);
                        po.setProvinceId(Long.valueOf(lineArr[6]));
                        po.setProvinceName(lineArr[7]);
                        poList.add(po);
                    }
                });
        return poList;
    }


}

